//package Hudi_SelfStudy
//
//import org.apache.hudi.DataSourceReadOptions.{BEGIN_INSTANTTIME, QUERY_TYPE_INCREMENTAL_OPT_VAL, QUERY_TYPE_OPT_KEY}
//import org.apache.spark.sql.SparkSession
//
//object 增量查询 {
//  def main(args: Array[String]): Unit = {

//    fileName:增量查询


//    /*
//          Hudi还提供了增量查询的方式，可以获取从给定提交时间戳以来更改的数据流。需要指定增量查询的beginTime，选择性指定endTime。如果我们希望在
//          给定提交之后进行所有更改，则不需要指定endTime（这是常见的情况）。
//     */
//  //  准备sparksql集成hudi的环境
//    val spark=SparkSession.builder()
//    .master("local[*]")
//    .appName("增量查询")
//    .config("spark.serializer","org.apache.spark.serializer.KryoSerializer")
//    .config("spark.sql.extensions","org.apache.spark.sql.hudi.HoodieSparkSessionExtension")
//    .enableHiveSupport()
//    .getOrCreate()
//
//    //   拿到表格数据并创建临时表
//    spark.read.format("hudi")
//      .load("hdfs://192.168.40.110:9000/user/hudi/hudi_trips_cow/")
//      .createOrReplaceTempView("temp01")
//
//    //   获取表格里面的插入时间    distinct(截然不同的):对数据进行去重
//    /*
//        .map( k => k.getString(0)):将查询结果的每一行转化为一个字符串，这里的0代表的是第一列的数据
//        .take(50):提取50哥元素
//     */
//     val time_list:List[String]=spark.sql(
//       """
//         |select
//         |distinct(_hoodie_commit_time) as insert_time
//         |from  temp01
//         |order by insert_time
//         |""".stripMargin).map( k => k.getString(0)).take(50)
//
//    //  获取指定的beginTime(因为这个列表里面就只有2个元素，所以-2)
//    val beginTime=time_list(time_list.length -2 )
//
//
//    //  增量加载数据，并创建临时表  increment:增量
//    //  里面的option是固定的格式，增量查询所需要的
//    val increment_query=spark.read.format("hudi")
//      .option(QUERY_TYPE_OPT_KEY,QUERY_TYPE_INCREMENTAL_OPT_VAL)
//      .option(BEGIN_INSTANTTIME,beginTime)
//      .load("hdfs://192.168.40.110:9000/user/hudi/hudi_trips_cow/")
//      .createOrReplaceTempView("temp02")
//
//
//
//    spark.close()
//
//  }
//
//}
